#include "../lib_include/EventLoop.h"
#include "../lib_include/Logger.h"
#include "../lib_include/Mutex.h"
#include "../lib_include/Channel.h"
#include "../lib_include/Poller.h"
#include "../lib_include/SocketsOps.h"
#include "../lib_include/TimerQueue.h"

#include <algorithm>
#include <signal.h>
#include <sys/eventfd.h>
#include <unistd.h>

using namespace netlib;
using namespace netlib::net;

namespace
{

__thread EventLoop* t_loopInThisThread = 0; // 每个线程只有一个EventLoop对象，线程间不共享

const int kPollTimeMs = 10000; // poll的超时时间

int createEventfd() // 创建eventfd
{
    int evtfd = ::eventfd(0, EFD_NONBLOCK | EFD_CLOEXEC);
    if (evtfd < 0)
    {
        LOG_SYSERR << "Failed in eventfd";
        abort();
    }
    return evtfd;
}

#pragma GCC diagnostic ignored "-Wold-style-cast"
class IgnoreSigPipe // 多线程程序应该屏蔽SIGPIPE信号
{
public:
    IgnoreSigPipe() { ::signal(SIGPIPE, SIG_IGN); }

}; // class IgnoreSigPipe end

#pragma GCC diagnostic error "-Wold-style-cast"

IgnoreSigPipe initObj; // 设置为全局变量对象，自动调用构造屏蔽SIGPIPE信号

} // namespace  end

EventLoop* EventLoop::getEventLoopOfCurrentThread() // 返回当前EventLoop对象的指针
{ return t_loopInThisThread; }

EventLoop::EventLoop()
  : looping_(false),
    quit_(false),
    eventHandling_(false),
    callingPendingFunctors_(false),
    iteration_(0),
    threadId_(CurrentThread::tid()),
    poller_(Poller::newDefaultPoller(this)),
    timerQueue_(new TimerQueue(this)),
    wakeupFd_(createEventfd()),
    wakeupChannel_(new Channel(this, wakeupFd_)),
    currentActiveChannel_(NULL)
{
        LOG_DEBUG << "EventLoop created " << this << " in thread " << threadId_;
        if (t_loopInThisThread)
        { // 如果当前线程已经创建，终止线程
            LOG_FATAL << "Another EventLoop " << t_loopInThisThread
                    << " exists in this thread " << threadId_;
        }
        else
            t_loopInThisThread = this;
        wakeupChannel_->setReadCallback(std::bind(&EventLoop::handleRead, this));

        wakeupChannel_->enableReading();
}

EventLoop::~EventLoop()
{
    LOG_DEBUG << "EventLoop " << this << " of thread " << threadId_
                << " destructs in thread " << CurrentThread::tid();
    wakeupChannel_->disableAll();
    wakeupChannel_->remove();
    ::close(wakeupFd_); // 关闭eventfd
    t_loopInThisThread = NULL;
}

void EventLoop::loop() // 事件循环函数，不能跨线程调用，只能由创建对象调用
{
        assert(!looping_);
        assertInLoopThread(); // 断言保证当前线程是创建者
        looping_ = true; // 开启事件循环
        quit_ = false;
        LOG_TRACE << "EventLoop " << this << " start looping"; // 记录日志

        while (!quit_)
        {
            activeChannels_.clear(); // 将Poller返回的活动通道

            // 返回时间戳和活动通道
            pollReturnTime_ = poller_->poll(kPollTimeMs, &activeChannels_);
            ++iteration_;
            if (Logger::logLevel() <= Logger::TRACE)
                printActiveChannels(); // 打印活动通道
            eventHandling_ = true; // 标志正在处理通道事件
            for (Channel* channel : activeChannels_) // 遍历所有的活动通道
            {
                currentActiveChannel_ = channel; // 设置当前通道为正在处理的状态
                currentActiveChannel_->handleEvent(pollReturnTime_); // 处理事件
            }
            currentActiveChannel_ = NULL; // 处理完毕，当前正处理的活动通道置空
            eventHandling_ = false; // 处理完毕，处于非处理通道事件状态
            doPendingFunctors(); // 运行附加任务
        }

        LOG_TRACE << "EventLoop " << this << " stop looping";
        looping_ = false;
}

void EventLoop::quit()
{
    quit_ = true; // 设置停止标志
    if (!isInLoopThread()) // 如果是跨线程，则需要唤醒EventLoop线程
        wakeup();
}

void EventLoop::runInLoop(Functor cb) // 执行传递过来的函数
{
    if (isInLoopThread()) // 如果在创建者线程中调用，则直接调用函数
        cb();
    else
        queueInLoop(std::move(cb)); // 添加至EventLoop附加任务列表中
}

void EventLoop::queueInLoop(Functor cb) // 向附加任务列表中添加任务
{
    {
        MutexLockGuard lock(mutex_);
        pendingFunctors_.push_back(std::move(cb)); // 向附加任务列表中添加任务
    }

    // 如果不是在EventLoop创建者线程中或者其正在处理附加任务时，需要唤醒该线程
    if (!isInLoopThread() || callingPendingFunctors_)
        wakeup();
}

size_t EventLoop::queueSize() const // 返回附加任务列表的长度
{
    MutexLockGuard lock(mutex_);
    return pendingFunctors_.size();
}

// 添加一个一次性定时器
TimerId EventLoop::runAt(Timestamp time, TimerCallback cb)
{ return timerQueue_->addTimer(std::move(cb), time, 0.0); }

// 添加一个一次性定时器，runAt的封装
TimerId EventLoop::runAfter(double delay, TimerCallback cb)
{
    Timestamp time(addTime(Timestamp::now(), delay));
    return runAt(time, std::move(cb));
}

// 添加一个重复的定时器
TimerId EventLoop::runEvery(double interval, TimerCallback cb)
{
    Timestamp time(addTime(Timestamp::now(), interval));
    return timerQueue_->addTimer(std::move(cb), time, interval);
}

void EventLoop::cancel(TimerId timerId) // 关闭定时器
{ return timerQueue_->cancel(timerId); }

void EventLoop::updateChannel(Channel* channel) // 更新poll事件表
{
    assert(channel->ownerLoop() == this);
    assertInLoopThread();
    poller_->updateChannel(channel); // 调用poll对象的updateChannel来更新事件表
}

void EventLoop::removeChannel(Channel* channel) // 在事件表中移除通道
{
    assert(channel->ownerLoop() == this);
    assertInLoopThread();
    if (eventHandling_)
    {
        assert(currentActiveChannel_ == channel ||
            std::find(activeChannels_.begin(), activeChannels_.end(), channel) == activeChannels_.end());
    }
    poller_->removeChannel(channel);
}

bool EventLoop::hasChannel(Channel* channel) // 查询该通道是否注册到poll中
{
    assert(channel->ownerLoop() == this); // 判断当前对象是否拥有该channel对象
    assertInLoopThread(); // 断言创建者线程
    return poller_->hasChannel(channel); // 判断该通道是否在poll参数列表中
}

void EventLoop::abortNotInLoopThread() // 出现跨线程调用函数，记录错误日志并退出
{
    LOG_FATAL << "EventLoop::abortNotInLoopThread - EventLoop " << this
                << " was created in threadId_ = " << threadId_
                << ", current thread id = " <<  CurrentThread::tid();
}

void EventLoop::wakeup() // 唤醒EventLoop所在的创建者线程
{
    uint64_t one = 1;
    
    // 向eventfd写入8个字节，poll触发可读事件，唤醒等待的线程
    ssize_t n = sockets::write(wakeupFd_, &one, sizeof one);
    if (n != sizeof one)
    {
        LOG_ERROR << "EventLoop::wakeup() writes " << n << " bytes instead of 8";
    }
}

void EventLoop::handleRead() // eventfd触发的处理函数，唤醒该线程
{
    uint64_t one = 1;
    ssize_t n = sockets::read(wakeupFd_, &one, sizeof one);
    if (n != sizeof one)
    {
        LOG_ERROR << "EventLoop::handleRead() reads " << n << " bytes instead of 8";
    }
}

void EventLoop::doPendingFunctors() // 运行附加任务
{
    std::vector<Functor> functors;
    callingPendingFunctors_ = true; // 设置状态为正在处理附加任务

    {
        MutexLockGuard lock(mutex_);
        functors.swap(pendingFunctors_);
    }
     // 如果此时pendingFunctors_有新加任务，则暂时搁置等待下一轮处理，防止IO线程阻塞于此
    for (const Functor& functor : functors)
        functor();
    callingPendingFunctors_ = false;
}

void EventLoop::printActiveChannels() const // 打印活动通道
{
    for (const Channel* channel : activeChannels_)
        LOG_TRACE << "{" << channel->reventsToString() << "} ";
}

